package com.hp.bon.sgw.service;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;

import javax.xml.transform.stream.StreamSource;
import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
import javax.xml.validation.Validator;

import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.transaction.annotation.Transactional;
import org.xml.sax.SAXException;

import com.google.gson.Gson;
import com.hp.bon.sgw.bean.XmlMessage;
import com.hp.bon.sgw.core.CallParam;
import com.hp.bon.sgw.domain.BusinessRecord;
import com.hp.bon.sgw.util.Comutils;
import com.hp.bon.sgw.util.dcc.DccAvpHelper;
import com.hp.ocs.diameter.message.Message;

public class SGWDataStoreService {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = LoggerFactory.getLogger(SGWDataStoreService.class);

	private SessionFactory sessionFact;

	/**
	 * 存放domain域对象
	 */
	private ConcurrentLinkedQueue<Object> dataQueue = new ConcurrentLinkedQueue<Object>();
	/**
	 * 保存需要进行状态更新的HSQL /SQL 语句
	 */
	private static final String HSQL_PREX = "H:";
	private ConcurrentLinkedQueue<String> updateSqlQueue = new ConcurrentLinkedQueue<String>();
	/**
	 * 保存以json格式输出的java对象，这些java对象会输出到指定的文件中，文件名是对象的类名
	 */
	private ConcurrentLinkedQueue<Object> jsonObjQueue = new ConcurrentLinkedQueue<Object>();

  private String jsonFileDir = "./data/json";

  private static final SchemaFactory schemaFactory = SchemaFactory.newInstance("http://www.w3.org/2001/XMLSchema");

  public ConcurrentLinkedQueue<Object> getDataQueue()
  {
    return this.dataQueue;
  }
  @Autowired
  @Qualifier("sessionFactory")
  public void setSessionFactory(SessionFactory sessionFactory) {
    this.sessionFact = sessionFactory;
  }

  public ConcurrentLinkedQueue<String> getUpdateSqlQueue()
  {
    return this.updateSqlQueue;
  }

  public void saveDomainData(Object domainObj) {
    this.dataQueue.offer(domainObj);
  }

  public boolean hasDomainDataToSave() {
    return !this.dataQueue.isEmpty();
  }

  public boolean hasSqlToExecute() {
    return !this.updateSqlQueue.isEmpty();
  }

  public boolean hasJsonToSave() {
    return !this.jsonObjQueue.isEmpty();
  }

  public void addHSQL(String theSql, boolean isHSQL) {
    String sql = isHSQL ? "H:" + theSql : theSql;
    this.updateSqlQueue.offer(sql);
  }

  public void addJsonJava(Object obj)
  {
    this.jsonObjQueue.offer(obj);
  }

  @Transactional("transactionManager")
  public boolean bulkExecSql(List<String> sqls)
  {
    Session session = null;
    try {
      session = this.sessionFact.getCurrentSession();
    } catch (Throwable e1) {
      logger.warn("can't connect db:" + e1);
      this.updateSqlQueue.clear();
      return false;
    }
    for (String sqlStr : sqls) {
      try {
        String hsql = sqlStr;
        int indx = sqlStr.indexOf("H:");
        boolean isHsql = false;
        if (indx == 0) {
          isHsql = true;
          hsql = sqlStr.substring("H:".length());
        }

        if (!isHsql)
          session.createSQLQuery(hsql).executeUpdate();
        else
          session.createQuery(hsql).executeUpdate();
      }
      catch (Exception e) {
        logger.warn("execute sql err:" + e + " sql:" + sqlStr);
      }
    }
    return true;
  }

	public void saveJsonFile() {
		if (jsonObjQueue.isEmpty()) {
			return;
		}
		Map<String, OutputStream> existStreamMap = new LinkedHashMap<String, OutputStream>();
		Iterator<Object> itor = jsonObjQueue.iterator();
		Gson gson = new Gson();
		byte[] lineBreakByts = { '\r', '\n' };
		while (itor.hasNext()) {
			try {
				Object data = itor.next();
				itor.remove();
				String clsName = data.getClass().getSimpleName();
				OutputStream outstream = getoutStream(clsName, existStreamMap);
				if (outstream != null) {
					outstream.write(gson.toJson(data).getBytes("UTF-8"));
					outstream.write(lineBreakByts);
				}
			} catch (Exception e) {
				logger.warn("write json err:" + e);
			}
		}
		// �ر��ļ�
		for (OutputStream out : existStreamMap.values()) {
			try {
				out.flush();
				out.close();
			} catch (IOException e) {
				logger.warn("close file err:" + e);
			}
		}
	}

  private OutputStream getoutStream(String objClsName, Map<String, OutputStream> existStreamMap)
  {
    String fileName = objClsName + "_" + Comutils.formatTime(System.currentTimeMillis(), "yyyyMMddHH") + ".json";
    OutputStream out = (OutputStream)existStreamMap.get(fileName);
    if (out == null) {
      try {
        File file = new File(this.jsonFileDir);
        if (!file.exists()) {
          file.mkdirs();
        }
        out = new BufferedOutputStream(new FileOutputStream(new File(this.jsonFileDir, fileName), true));
        existStreamMap.put(fileName, out);
      } catch (FileNotFoundException e) {
        logger.warn("create out file err:" + e);
      }
    }
    return out;
  }

  public void setJsonFileDir(String jsonFileDir) {
    this.jsonFileDir = jsonFileDir;
  }

  public String getJsonFileDir() {
    return this.jsonFileDir;
  }

  public ConcurrentLinkedQueue<Object> getJsonObjQueue() {
    return this.jsonObjQueue;
  }

  public static BusinessRecord convertCallParamToBusinessRecord(CallParam callParam)
  {
    if (callParam.getFromNode() != null) {
      BusinessRecord brec = new BusinessRecord();
      brec.setBusinessCallId(callParam.getBussinessCallId());
      brec.setServiceContextId(callParam.getServiceContextId());
      brec.setSessionId(callParam.getSessionId());
      brec.setProtocolType(callParam.getAsynHandler().getProtocolType());
      brec.setRequestHost(callParam.getUserName());
      brec.setResult(callParam.getRetCode());
      brec.setResultReason(callParam.getResultMsg());
      brec.setFromNodeId(callParam.getFromNode().getId());

      if (callParam.getFromNode().getNodeType() == 1)
        brec.setTransDirection(0);
      else {
        brec.setTransDirection(1);
      }
      if (callParam.getToNode() != null) {
        brec.setToNodeId(callParam.getToNode().getId());
      }

      if (callParam.getRequest() != null) {
        brec.setRecvRequestTime(callParam.getRecvReqTime());

        if ((callParam.getRequest() instanceof XmlMessage)) {
          XmlMessage msg = (XmlMessage)callParam.getRequest();
          brec.setRecvRequestLength(msg.getLength());

          brec.setRequestIP(msg.getIp());
        } else if ((callParam.getRequest() instanceof Message)) {
          Message msg = (Message)callParam.getRequest();
          brec.setRecvRequestLength(msg.header().messageLength());
        }
      }

      if (callParam.getSendRequest() != null) {
        brec.setSendRequestTime(callParam.getSendReqTime());

        if ((callParam.getSendRequest() instanceof XmlMessage)) {
          XmlMessage msg = (XmlMessage)callParam.getSendRequest();
          brec.setSendRequestLength(msg.getLength());
        } else if ((callParam.getSendRequest() instanceof Message)) {
          Message msg = (Message)callParam.getSendRequest();
          brec.setSendRequestLength(msg.header().messageLength());
        }
      }

      if (callParam.getRecvResponse() != null) {
        brec.setRecvResponseTime(callParam.getRecvRespTime());

        if ((callParam.getRecvResponse() instanceof XmlMessage)) {
          XmlMessage msg = (XmlMessage)callParam.getRecvResponse();
          brec.setRecvResponseLength(msg.getLength());
          brec.setResponseHost(msg.getUID());
        } else if ((callParam.getRecvResponse() instanceof Message)) {
          Message msg = (Message)callParam.getRecvResponse();
          brec.setRecvResponseLength(msg.header().messageLength());
          DccAvpHelper.processCCAMessgeInfos(msg, brec);
        }
      }

      if (callParam.getResponse() != null) {
        brec.setSendResponseTime(callParam.getSendRespTime());

        if ((callParam.getResponse() instanceof XmlMessage)) {
          XmlMessage msg = (XmlMessage)callParam.getResponse();
          brec.setSendResponseLength(msg.getLength());

          if (callParam.getRetCode() == 0)
            try {
              StreamSource ss = (StreamSource)ConfigService.answerValidatorMap.get(callParam.getCapability());
              if (ss != null) {
                if (msg.getDATA() != null) {
                  InputStream inputStream = new ByteArrayInputStream(msg.getDATA().getBytes());
                  Schema schema = schemaFactory.newSchema(ss);
                  Validator val = schema.newValidator();
                  val.validate(new StreamSource(inputStream));
                  brec.setAnswerValidateCode(Integer.valueOf(0));
                } else {
                  brec.setValidateDesc("ACK DATA is null.");
                }
              }
              else brec.setValidateDesc("no pattern,not need validator"); 
            }
            catch (SAXException e)
            {
              brec.setAnswerValidateCode(Integer.valueOf(1));
              brec.setValidateDesc(e.getMessage().substring(e.getMessage().indexOf(" ")));
            }
            catch (Exception localException)
            {
            }
        }
        else if ((callParam.getResponse() instanceof Message)) {
          Message msg = (Message)callParam.getResponse();
          brec.setSendResponseLength(msg.header().messageLength());
          if (callParam.getRecvResponse() == null) {
            DccAvpHelper.processCCAMessgeInfos(msg, brec);
          }
        }
      }
      brec.setResult(callParam.getRetCode());
      brec.setResultReason(callParam.getResultMsg());
      brec.setServiceResultCode(Integer.valueOf(callParam.getServiceResultCode()));
      return brec;
    }
    return null;
  }

  @Transactional("transactionManager")
  public void updateChannelStatus2DB(int channelId, int status) {
    try {
      StringBuilder sb = new StringBuilder();
      sb.append("update SGWChannel set status =").append(status);
      sb.append(",statusTime = sysDate");
      sb.append(" where id= ").append(channelId);
      addHSQL(sb.toString(), true);
    }
    catch (Exception e) {
      logger.warn("update channel status to DB failed :channelId=" + channelId + "," + "status=" + status + "," + e.toString());
    }
  }

  @Transactional("transactionManager")
  public void updateNodeStatus2DB(int nodeId, int status) {
    try {
      StringBuilder sb = new StringBuilder();
      sb.append("update Node set status =").append(status);
      sb.append(",statusTime = sysDate");
      sb.append(" where id= ").append(status);
      addHSQL(sb.toString(), true);
    }
    catch (Exception e) {
      logger.warn("update node status to DB failed :nodeId=" + nodeId + "," + "status=" + status + "," + e.toString());
    }
  }
}